package de.tuberlin.dima.aim3.oc;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.commons.lang.builder.EqualsBuilder;
import org.apache.commons.lang.builder.HashCodeBuilder;

import de.tuberlin.dima.aim3.oc.input.custom.WikiPage;
import de.tuberlin.dima.aim3.oc.input.custom.WikiRevision;
import de.tuberlin.dima.aim3.oc.input.custom.WikiUser;
import eu.stratosphere.pact.common.type.Value;

/**
 * Serializable wrapper for {@link WikiPage} instances.
 * 
 * @author Florian Feigenbutz <florian.feigenbutz@campus.tu-berlin.de>
 * 
 */
public class PactWikiPage implements Value {

  private WikiPage page;

  public PactWikiPage(WikiPage page) {
    super();
    this.page = page;
  }

  @Override
  public void write(DataOutput out) throws IOException {
    out.writeUTF(page.getId());
    out.writeUTF(page.getTitle());
    out.writeInt(page.getRevisions().size());
    for (WikiRevision revision : page.getRevisions()) {
      out.writeUTF(revision.getId());
      out.writeUTF(revision.getTimestamp());
      out.writeBoolean(revision.getMinorChange());
      out.writeUTF(revision.getComment());
      out.writeUTF(revision.getText());
      out.writeUTF(revision.getUser().getUserid());
      out.writeUTF(revision.getUser().getUserip());
      out.writeUTF(revision.getUser().getUsername());
    }
  }

  @Override
  public void read(DataInput in) throws IOException {
    String pageId = in.readUTF();
    String pageTitle = in.readUTF();
    int numberOfRevisions = in.readInt();
    List<WikiRevision> revisions = new ArrayList<WikiRevision>(
        numberOfRevisions);
    for (int i = 0; i < numberOfRevisions; i++) {
      String revisionId = in.readUTF();
      String revisionTimestamp = in.readUTF();
      boolean revisionMinorChange = in.readBoolean();
      String revisionComment = in.readUTF();
      String revisionText = in.readUTF();

      String userid = in.readUTF();
      String userip = in.readUTF();
      String username = in.readUTF();
      WikiUser user = new WikiUser(username, userid, userip);

      WikiRevision revision = new WikiRevision(revisionId, revisionTimestamp,
          user, revisionMinorChange, revisionComment, revisionText);

      revisions.add(revision);
    }
    page = new WikiPage(pageId, pageTitle, revisions);
  }

  @Override
  public int hashCode() {
    return HashCodeBuilder.reflectionHashCode(this);
  }

  @Override
  public boolean equals(Object obj) {
    return EqualsBuilder.reflectionEquals(obj, this);
  }

  @Override
  public String toString() {
    return page != null ? page.toString() : "";
  }

}
